package cn.monkey.transport.netty.server;

import cn.monkey.transport.netty.ShutDown;
import cn.monkey.transport.core.server.Server;
import cn.monkey.transport.netty.Transport;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.socket.SocketChannel;
import io.netty.util.AttributeKey;
import io.netty.util.concurrent.Future;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.net.SocketAddress;
import java.util.*;
import java.util.function.Consumer;
import java.util.function.Supplier;

public class ServerTransport extends Transport<ServerBootstrap, ServerTransportConfig, ServerTransportConfig.Builder, ServerChannel>
        implements Server {
    private static final Logger log = LoggerFactory.getLogger(ServerTransport.class);

    public ServerTransport(ServerTransportConfig config) {
        super(config);
    }

    @Override
    protected ServerTransportConfig fixedConfig(ServerTransportConfig config) {
        ShutDown transportShutDown = config.shutDown();
        if (transportShutDown == null) {
            return config.mutate().shutDown(transport -> {
                List<Mono<Void>> monoList = this.channelMap.values()
                        .stream()
                        .map(ChannelOutboundInvoker::close)
                        .map(channelFuture -> Mono.create(sink -> channelFuture.addListener((ChannelFutureListener) future -> {
                            if (future.isSuccess()) {
                                sink.success();
                                return;
                            }
                            sink.error(future.cause());
                        })).then()).toList();
                ArrayList<Mono<Void>> monos = new ArrayList<>(monoList);
                Future<Object> eventLoopGroupShutDownFuture = this.shutDownEventLoopGroup(config.eventLoopGroup());
                Future<Object> childLoopGroupShutDownFuture = this.shutDownEventLoopGroup(config.childGroup());
                monos.add(Mono.create(sink -> childLoopGroupShutDownFuture.addListener(future -> {
                    if (future.isSuccess()) {
                        sink.success();
                        return;
                    }
                    sink.error(future.cause());
                })));
                monos.add(Mono.create(sink -> eventLoopGroupShutDownFuture.addListener(future -> {
                    if (future.isSuccess()) {
                        sink.success();
                        return;
                    }
                    sink.error(future.cause());
                })));
                return Flux.merge(monos).then();
            }).build();
        }
        return config;
    }

    protected Mono<Void> start0(ServerBootstrap bootstrap) {
        List<Supplier<SocketAddress>> addresses = this.config.socketAddresses();
        if (addresses == null || addresses.isEmpty()) {
            throw new NullPointerException("empty socketAddress");
        }
        List<Mono<Void>> monoList = addresses.stream()
                .map(Supplier::get)
                .filter(Objects::nonNull)
                .map(socketAddress -> {
                    ChannelFuture channelFuture = bootstrap.bind(socketAddress);
                    Channel channel = channelFuture.channel();
                    return Mono.create(sink -> channelFuture.addListener((ChannelFutureListener) future -> {
                        if (!future.isSuccess()) {
                            sink.error(future.cause());
                            return;
                        }
                        if (channel instanceof ServerChannel serverChannel) {
                            log.info("socket: {} open success", socketAddress);
                            this.channelMap.put(socketAddress, serverChannel);
                            sink.success();
                        }
                    })).then();
                }).toList();
        return Flux.merge(monoList).then();
    }

    protected void bindChildChannelAttributes(SocketChannel socketChannel) {
        Map<AttributeKey<?>, ?> channelAttributes = this.config.childChannelAttributes();
        if (channelAttributes != null && !channelAttributes.isEmpty()) {
            for (Map.Entry<AttributeKey<?>, ?> e : channelAttributes.entrySet()) {
                @SuppressWarnings("unchecked") AttributeKey<Object> key = (AttributeKey<Object>) e.getKey();
                socketChannel.attr(key).set(e.getValue());
            }
        }
    }

    @Override
    protected ServerBootstrap doOnChannelInit(ServerBootstrap bootstrap) {
        Consumer<Channel> channelCustomizer = this.config.channelInitCustomizer();
        return bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) {
                ServerTransport.this.bindChildChannelAttributes(ch);
                if (channelCustomizer != null) {
                    channelCustomizer.accept(ch);
                }
            }
        });
    }

    @Override
    protected ServerBootstrap bindEventLoopGroup(ServerBootstrap bootstrap) {
        return bootstrap.group(this.config.eventLoopGroup(), this.config.childGroup());
    }

    protected ServerBootstrap newBootStrap() {
        return new ServerBootstrap();
    }

    @Override
    public Mono<Void> start() {
        return this.start0(this.bootstrap);
    }

    @Override
    public Mono<Void> stop() {
        return this.config.shutDown().apply(this);
    }
}
